package com.lzq.mq.broker.event.spi.listener;

import com.alibaba.fastjson.JSON;
import com.lzq.mq.broker.cache.CommonCache;
import com.lzq.mq.broker.event.model.CreateTopicEvent;
import com.lzq.mq.broker.model.CommitLogModel;
import com.lzq.mq.broker.model.ConsumeQueueModel;
import com.lzq.mq.broker.model.MqTopicModel;
import com.lzq.mq.broker.utils.FileNameUtil;
import com.lzq.mq.common.coder.TcpMsg;
import com.lzq.mq.common.constants.BrokerConstants;
import com.lzq.mq.common.dto.CreateTopicReqDTO;
import com.lzq.mq.common.dto.CreateTopicRespDTO;
import com.lzq.mq.common.enums.BrokerResponseCode;
import com.lzq.mq.common.event.Listener;
import com.lzq.mq.common.utils.AssertUtils;
import lombok.extern.slf4j.Slf4j;

import java.io.File;
import java.io.IOException;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.atomic.AtomicInteger;
/**
 * 创建topic监听器
 */
@Slf4j
public class CreateTopicListener implements Listener<CreateTopicEvent> {


    @Override
    public void onReceive(CreateTopicEvent event) throws Exception {
        CreateTopicReqDTO dto = event.getCreateTopicReqDTO();
        AssertUtils.isTrue(dto.getQueueSize() > 0 && dto.getQueueSize() < 100, "queueSize参数异常");
        MqTopicModel eagleMqTopicModel = CommonCache.getMqTopicModelMap().get(dto.getTopic());
        AssertUtils.isTrue(eagleMqTopicModel == null, "topic已经存在");
        createTopicFile(dto);
        addTopicInCommonCache(dto);
        loadFileInMMap(dto);
        log.info("topic:{} 创建成功! 队列数量: {}", dto.getTopic(), dto.getQueueSize());

        // 返回操作结果
        CreateTopicRespDTO createTopicRespDTO = new CreateTopicRespDTO();
        createTopicRespDTO.setMsgId(dto.getMsgId());
        createTopicRespDTO.setTopic(dto.getTopic());
        createTopicRespDTO.setQueueSize(dto.getQueueSize());
        createTopicRespDTO.setSuccess(true);
        TcpMsg tcpMsg = new TcpMsg(BrokerResponseCode.CREATED_TOPIC_SUCCESS.getCode(), JSON.toJSONBytes(createTopicRespDTO));
        event.getChannelHandlerContext().writeAndFlush(tcpMsg);
    }

    /**
     * 创建topic对应的CommitLog和ConsumeQueue文件
     */
    public static void createTopicFile(CreateTopicReqDTO dto) throws IOException {

        // commitlog
        String baseCLDir = FileNameUtil.buildCommitLogBasePath(dto.getTopic());
        File newCLDirFile = new File(baseCLDir);
        newCLDirFile.mkdir();       // 创建文件夹
        File commitLogFile = new File(baseCLDir + BrokerConstants.SPLIT + FileNameUtil.buildFirstCommitLogFileName());
        commitLogFile.createNewFile();

        // consume queue
        String baseConsumeQueueDirPath = FileNameUtil.buildConsumeQueueBasePath(dto.getTopic());
        File consumeQueueDir = new File(baseConsumeQueueDirPath);
        consumeQueueDir.mkdir();
        for (int i = 0; i < dto.getQueueSize(); i++) {
            new File(baseConsumeQueueDirPath + BrokerConstants.SPLIT + i).mkdir();
            new File(baseConsumeQueueDirPath + BrokerConstants.SPLIT + i + BrokerConstants.SPLIT + FileNameUtil.buildFirstConsumeQueueName())
                    .createNewFile();
        }
    }

    /**
     * 将文件映射到mmap中
     */
    public static void loadFileInMMap(CreateTopicReqDTO createTopicReqDTO) {
        CommonCache.getCommitLogAppendHandler().prepareMMapLoading(createTopicReqDTO.getTopic());
        CommonCache.getConsumeQueueAppendHandler().prepareConsumeQueue(createTopicReqDTO.getTopic());
    }

    /**
     * 添加topic到缓存中
     */
    public static void addTopicInCommonCache(CreateTopicReqDTO createTopicReqDTO) {
        MqTopicModel mqTopicModel = new MqTopicModel();
        mqTopicModel.setTopicName(createTopicReqDTO.getTopic());

        long currentTimeStamp = System.currentTimeMillis();
        mqTopicModel.setCreateAt(currentTimeStamp);
        mqTopicModel.setUpdateAt(currentTimeStamp);

        CommitLogModel commitLogModel = new CommitLogModel();
        commitLogModel.setFileName(FileNameUtil.buildFirstCommitLogFileName());
        commitLogModel.setOffsetLimit(BrokerConstants.COMMITLOG_DEFAULT_MMAP_SIZE.longValue());
        commitLogModel.setOffset(new AtomicInteger(0));
        mqTopicModel.setCommitLogModel(commitLogModel);

        List<ConsumeQueueModel> queueList = new ArrayList<>();
        for (int i = 0; i < createTopicReqDTO.getQueueSize(); i++) {
            ConsumeQueueModel queueModel = new ConsumeQueueModel();
            queueModel.setId(i);
            queueModel.setFileName(FileNameUtil.buildFirstConsumeQueueName());
            queueModel.setOffsetLimit(BrokerConstants.COMSUMEQUEUE_DEFAULT_MMAP_SIZE);
            queueModel.setLastOffset(0);
            queueModel.setLatestOffset(new AtomicInteger(0));
            queueList.add(queueModel);
        }
        mqTopicModel.setQueueList(queueList);
        CommonCache.getMqTopicModelList().add(mqTopicModel);
    }
}
